package com.nepu.gmall.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.nepu.gmall.realtime.util.DimUtil;
import com.nepu.gmall.realtime.util.DruidDSUtil;
import com.nepu.gmall.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import javax.sound.midi.Soundbank;
import java.lang.reflect.InvocationTargetException;
import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author chenshuaijun
 * @create 2023-03-06 10:15
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimAsyncInterface<T>{
    private DruidDataSource dataSource;
    private ThreadPoolExecutor threadPoolExecutor;

    private String tableName;

    public DimAsyncFunction() {
    }

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidDSUtil.createDataSource();
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {

        threadPoolExecutor.execute(() -> {
            try {
                // 创建Phoenix的连接
                DruidPooledConnection connection = dataSource.getConnection();
                // 拿到需要进行join的key
                String key = getKey(input);
                // 查询维度表
                JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, key);
                // 归还连接
                connection.close();
                // 封装信息
                encapsulateBeans(input, dimInfo);
                // 将数据写出
                resultFuture.complete(Collections.singleton(input));
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println(input+" 在关联维表"+tableName+" 的时候出现了维表关联失败的问题");
            }

        });

    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println(input+"关联维表超时");
    }
}
